package com.raven.play.datalink.common.type;

import org.apache.avro.LogicalTypes;
import org.apache.avro.Schema;
import org.apache.avro.SchemaBuilder;

import java.util.List;

public class RecordHelper {

    public static Schema createSchema(String name, List<? extends SimpleField> fields) {
        SchemaBuilder.FieldAssembler<Schema> fieldAssembler = SchemaBuilder.record(name).fields();
        for (SimpleField field : fields) {
            assembly(fieldAssembler, field);
        }
        return fieldAssembler.endRecord();
    }

    private static SchemaBuilder.FieldAssembler<Schema> assembly(SchemaBuilder.FieldAssembler<Schema> fieldAssembler, SimpleField field) {
        Schema fieldSchema = getSchemaForType(field);
        if (field.isNullable()) {
            fieldSchema = Schema.createUnion(Schema.create(Schema.Type.NULL), fieldSchema);
        }
        return fieldAssembler.name(field.getName())
                .doc(field.getDoc())
                .type(fieldSchema)
                .withDefault(field.getDefaultValue());
    }

    private static Schema getSchemaForType(SimpleField field) {
        TypeEnum type = field.getType();
        switch (type) {
            case STRING:
                return Schema.create(Schema.Type.STRING);
            case BOOLEAN:
                return Schema.create(Schema.Type.BOOLEAN);
            case BYTE:
                return Schema.create(Schema.Type.BYTES);
            case INTEGER:
                return Schema.create(Schema.Type.INT);
            case LONG:
                return Schema.create(Schema.Type.LONG);
            case FLOAT:
                return Schema.create(Schema.Type.FLOAT);
            case DOUBLE:
                return Schema.create(Schema.Type.DOUBLE);

            case ARRAY:
                return Schema.create(Schema.Type.ARRAY);
            case MAP:
                return Schema.create(Schema.Type.MAP);
            case ENUM:
                return Schema.create(Schema.Type.ENUM);

            case BIG_DECIMAL:
                BigDecimalField bigDecimalField = (BigDecimalField) field;
                return LogicalTypes.decimal(bigDecimalField.getPrecision(), bigDecimalField.getScale()).addToSchema(Schema.create(Schema.Type.BYTES));
            case DATE:
                return LogicalTypes.date().addToSchema(Schema.create(Schema.Type.BYTES));
            case TIME:
                TimePrecisionEnum timePrecision = ((TimeField) field).getPrecision();
                if (TimePrecisionEnum.MILLIS.equals(timePrecision)) {
                    return LogicalTypes.timeMillis().addToSchema(Schema.create(Schema.Type.BYTES));
                } else if (TimePrecisionEnum.MICROS.equals(timePrecision)) {
                    return LogicalTypes.timeMicros().addToSchema(Schema.create(Schema.Type.BYTES));
                } else {
                    throw new IllegalArgumentException("Unsupported time precision: " + timePrecision);
                }
            case DATETIME:
                TimePrecisionEnum timestampPrecision = ((DatetimeField) field).getPrecision();
                if (TimePrecisionEnum.MILLIS.equals(timestampPrecision)) {
                    return LogicalTypes.timestampMillis().addToSchema(Schema.create(Schema.Type.BYTES));
                } else if (TimePrecisionEnum.MICROS.equals(timestampPrecision)) {
                    return LogicalTypes.timestampMicros().addToSchema(Schema.create(Schema.Type.BYTES));
                } else {
                    throw new IllegalArgumentException("Unsupported time precision: " + timestampPrecision);
                }

            default:
                throw new IllegalArgumentException("Unsupported type: " + type);
        }
    }
}
